package com.kafka.service;


import cn.hutool.db.Entity;
import com.kafka.utils.DBUtils;

import java.sql.SQLException;

/**
 * @Author: Jiangxx
 * @Date: 2022/01/12
 * @Description:
 */
public class OffsetService {

    private static final String INSERT_SQL = "insert into offset_information(`topic`,`partition_name`,`partition_num`,`offset`) values(?,?,?,?)";
    private static final String SELECT_SQL = "select `topic`,`partition_name`,`partition_num`,`offset` from offset_information where topic = ? and partition_num = ? order by create_time desc limit 1";
    private static final String OFFSET = "offset";

    public void saveOffset(String topic, String partitionName, int partitionNum, long offset){
        String TABLE_NAME = "offset_information";
        String OFFSET = "offset";
        String TOPIC = "topic";
        String PARTITION_NUM = "partition_num";
        String PARTITION_NAME = "partition_name";
        try {
            long count = DBUtils.getDB().count(Entity.create(TABLE_NAME)
                    .set(TOPIC, topic)
                    .set(PARTITION_NUM, partitionNum)
                    .set(PARTITION_NAME, partitionName));
            if (count > 0) {
                DBUtils.getDB().update(Entity.create(TABLE_NAME)
                                .set(OFFSET, offset),
                        Entity.create(TABLE_NAME)
                                .set(TOPIC, topic)
                                .set(PARTITION_NUM, partitionNum)
                                .set(PARTITION_NAME, partitionName));
            } else {
                DBUtils.getDB().insert(Entity.create(TABLE_NAME)
                        .set(TOPIC, topic)
                        .set(PARTITION_NUM, partitionNum)
                        .set(PARTITION_NAME, partitionName)
                        .set(OFFSET, offset));
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public Long getOffset(int partitionNum, String topic) {
        Entity entity = null;
        try {
            entity = DBUtils.getDB().get(Entity.create("offset_information")
                    .set("topic", topic)
                    .set("partition_num", partitionNum));
        } catch (SQLException e) {
            e.printStackTrace();
        }
        if (entity == null) {
            return null;
        }
        return (Long) entity.get("offset");
    }
}
